package com.apes.framework.plugin.outbox;

import com.apes.framework.api.DomainService;
import com.apes.framework.api.SpringManager;
import com.apes.framework.plugin.cache.Cache;
import com.apes.framework.plugin.cache.MemoryCache;
import com.apes.framework.rop.session.SessionUtil;
import com.apes.framework.util.MapUtil;
import com.apes.framework.util.Tools;
import org.springframework.context.ApplicationContext;
import org.springframework.core.annotation.AnnotationUtils;
import org.springframework.stereotype.Component;
import org.springframework.util.ReflectionUtils;
import org.springframework.util.StringUtils;

import java.util.*;
import java.util.stream.Collectors;

import static java.util.stream.Collectors.groupingBy;

/**
 * 功能：
 *
 * @author xul
 * @create 2019-10-26 14:14
 */
@Component
public class OutboxSubscriberManager extends DomainService {
    private final String CONST_DEFAULT_CACHENAME = "outboxSubscriberCache";
    private final String CONST_DEFAULT_CACHENAME_INIT_FLAG = "outboxSubscriberCacheInitFlag";
    private boolean localInit = false;
    private List<OutboxSubscriber> localOutboxSubscribers;

    //@Autowired
    private Cache cache = MemoryCache.getInstance();

    public void addOutboxSubscriber(OutboxSubscriber subscriber) {
        List<OutboxSubscriber> outboxSubscribers = getOutboxSubscribers(subscriber.getEventType(), subscriber.getAggregateType());
        if (outboxSubscribers == null) outboxSubscribers = new ArrayList<>();
        outboxSubscribers.removeIf(outboxSubscriber -> outboxSubscriber.getId().equals(subscriber.getId()));
        outboxSubscribers.add(subscriber);
        addOutboxSubscribers(subscriber.getEventType(), subscriber.getAggregateType(), outboxSubscribers);
    }

    public void removeOutboxSubscriber(OutboxSubscriber subscriber) {
        List<OutboxSubscriber> outboxSubscribers = getOutboxSubscribers(subscriber.getEventType(), subscriber.getAggregateType());
        if (outboxSubscribers == null) return;
        outboxSubscribers.removeIf(outboxSubscriber -> outboxSubscriber.getId().equals(subscriber.getId()));
        if (outboxSubscribers.isEmpty()) {
            String id = SessionUtil.getSessionId(subscriber.getEventType(), subscriber.getAggregateType());
            cache.remove(CONST_DEFAULT_CACHENAME, id);
            return;
        }
        addOutboxSubscribers(subscriber.getEventType(), subscriber.getAggregateType(), outboxSubscribers);
    }

    public void addOutboxSubscribers(String eventType, String aggregateType, List<OutboxSubscriber> outboxSubscribers) {
        String id = SessionUtil.getSessionId(eventType, aggregateType);
        cache.put(CONST_DEFAULT_CACHENAME, id, outboxSubscribers);
    }

    public OutboxSubscriber getOutboxSubscriber(String outboxSubscriberId, String eventType, String aggregateType) {
        List<OutboxSubscriber> outboxSubscribers = getOutboxSubscribers(eventType, aggregateType);
        if (outboxSubscribers == null) return null;
        return outboxSubscribers.stream().filter(outboxSubscriber -> outboxSubscriber.getId().equals(outboxSubscriberId)).findAny().orElse(null);
    }

    public List<OutboxSubscriber> getOutboxSubscribers(String eventType, String aggregateType) {
        Boolean init = (Boolean) cache.get(CONST_DEFAULT_CACHENAME, CONST_DEFAULT_CACHENAME_INIT_FLAG);
        if (init == null || !init) {
            List<OutboxSubscriber> outboxSubscribers = getOutboxSubscribers();
            outboxSubscribers.stream().collect(groupingBy(OutboxSubscriber::cacheId))
                    .forEach((k, v) -> cache.put(CONST_DEFAULT_CACHENAME, k, v));
            cache.put(CONST_DEFAULT_CACHENAME, CONST_DEFAULT_CACHENAME_INIT_FLAG, true);
        }
        String id = SessionUtil.getSessionId(eventType, aggregateType);
        List<OutboxSubscriber> outboxSubscribers = new ArrayList<>();
        List<OutboxSubscriber> subscribers = (List<OutboxSubscriber>) cache.get(CONST_DEFAULT_CACHENAME, id);
        if (subscribers != null) {
            outboxSubscribers.addAll(subscribers);
        }
        if (getLocalOutboxSubscribers() != null) {
            outboxSubscribers.addAll(localOutboxSubscribers.stream()
                    .filter(subscriber -> subscriber != null && eventType.equals(subscriber.getEventType()) && aggregateType.equals(subscriber.getAggregateType()))
                    .collect(Collectors.toList()));
        }
        return outboxSubscribers.stream().distinct().collect(Collectors.toList());
    }

    public List<OutboxSubscriber> getOutboxSubscribers() {
        List<OutboxSubscriber> outboxSubscribers = new ArrayList<>();
        List result = sendSynEvent("outbox.subscriber.init", "{}");
        merge(outboxSubscribers, result);
        return outboxSubscribers.stream().distinct().collect(Collectors.toList());
    }

    public OutboxSubscriber createOutboxSubscriber(Class clazz, String outboxSubscriberId, String outboxSubscriberName, String eventType, String consumer) {
        OutboxSubscriber outboxSubscriber = new OutboxSubscriber();
        outboxSubscriber.setId(outboxSubscriberId);
        outboxSubscriber.setEventType(eventType);
        outboxSubscriber.setAggregateType(clazz.getName());
        outboxSubscriber.setName(outboxSubscriberName);
        outboxSubscriber.setConsumer(consumer);
        outboxSubscriber.setReadonly(true);
        return outboxSubscriber;
    }

    public synchronized List<OutboxSubscriber> getLocalOutboxSubscribers() {
        if (!localInit) {
            localInit = true;
            try {
                localOutboxSubscribers = new ArrayList<>();
                ApplicationContext context = SpringManager.getApplicationContext();
                addLocalOutboxSubscribers(context);
                List result = sendSynEvent("outbox.subscriber.local.init", "{}");
                merge(localOutboxSubscribers, result);
            } catch (Exception e) {
                localInit = false;
            }
            localOutboxSubscribers = localOutboxSubscribers.stream().distinct().collect(Collectors.toList());
        }
        return localOutboxSubscribers;
    }

    private void merge(List<OutboxSubscriber> outboxSubscribers, List result) {
        if (result == null || result.isEmpty()) return;
        result.stream().forEach(m -> {
            if (m instanceof Collection) {
                outboxSubscribers.addAll((Collection<? extends OutboxSubscriber>) m);
            } else if (m instanceof OutboxSubscriber) {
                outboxSubscribers.add((OutboxSubscriber) m);
            }
        });
    }

    private void addLocalOutboxSubscribers(final ApplicationContext context) {
        String[] beanNames = context.getBeanNamesForAnnotation(Component.class);
        for (final String beanName : beanNames) {
            Class<?> handlerType = context.getType(beanName);
            ReflectionUtils.doWithMethods(handlerType, method -> {
                ReflectionUtils.makeAccessible(method);
                com.apes.framework.plugin.outbox.annotation.OutboxSubscriber annotation = AnnotationUtils.findAnnotation(method, com.apes.framework.plugin.outbox.annotation.OutboxSubscriber.class);
                OutboxSubscriber subscriber = new OutboxSubscriber();
                String id = annotation.id();
                if (StringUtils.isEmpty(id)) {
                    id = String.format("local.%s.%s", handlerType.getSimpleName(), method.getName());
                }
                subscriber.setId(id);
                subscriber.setName(annotation.name());
                subscriber.setEventType(annotation.eventType());
                String condition = annotation.condition();
                if (StringUtils.hasText(condition)) {
                    condition = Tools.replace(condition);
                }
                subscriber.setCondition(condition);
                String aggregateType = annotation.aggregateType();
                if (StringUtils.isEmpty(aggregateType)) {
                    aggregateType = Arrays.stream(method.getParameterTypes()).findFirst().get().getName();
                }
                subscriber.setAggregateType(aggregateType);
                subscriber.setProcessor(o -> {
                    Object target = context.getBean(beanName);
                    return ReflectionUtils.invokeMethod(method, target, o);
                });
                localOutboxSubscribers.add(subscriber);
            }, method -> !method.isSynthetic() && method.getParameterTypes().length == 1
                    && AnnotationUtils.findAnnotation(method, com.apes.framework.plugin.outbox.annotation.OutboxSubscriber.class) != null);
        }

        if (context.getParent() != null) {
            addLocalOutboxSubscribers(context.getParent());
        }
    }

    public List<OutboxSubscriber> getAllOutboxSubscribers() {
        List<OutboxSubscriber> outboxSubscribers = getOutboxSubscribers();
        if (outboxSubscribers == null) outboxSubscribers = new ArrayList<>();
        outboxSubscribers.addAll(getLocalOutboxSubscribers());
        outboxSubscribers = outboxSubscribers.stream().distinct().collect(Collectors.toList());
        outboxSubscribers.sort(Comparator.comparing(OutboxSubscriber::getId));
        return outboxSubscribers;
    }
}
